package cn.springboot.linked;

import java.io.IOException;

import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.analysis.TokenStream;
import org.apache.lucene.analysis.cn.smart.SmartChineseAnalyzer;
import org.apache.lucene.analysis.tokenattributes.CharTermAttribute;

/**
 * @author liufl E-mail:Huberier@allpyra.com
 * @version 创建时间：2017年9月27日 下午4:33:10
 * 在Map阶段获得文件名，并将文件名还原为url，作为value。将文件内容通过分词器分词后将分出的每个词作为key，输出。 
 * 将用户数据的关键字分词后与倒排索引分别匹配，只要匹配到的在Combiner中统计词频，并在Reduce中操作后输出。 
 */
public class InvertedIndexMapper extends Mapper<Object, Text, Text, Text>{

    private Text keyInfo = new Text();  // 存储单词和URI的组合
    private Text valueInfo = new Text(); //存储词频
    private FileSplit split;  // 存储split对象。

    @Override
    protected void map(Object key, Text value, Mapper<Object, Text, Text, Text>.Context context)
            throws IOException, InterruptedException {

        //获得<key,value>对所属的FileSplit对象。
        split = (FileSplit) context.getInputSplit();

        Analyzer sca = new SmartChineseAnalyzer( );  

        TokenStream ts = sca.tokenStream("field", value.toString());  
        CharTermAttribute ch = ts.addAttribute(CharTermAttribute.class);  

        ts.reset();  
        while (ts.incrementToken()) {  
            System.out.println(ch.toString());  
            String url=split.getPath().toString();
            url=url.substring(url.lastIndexOf("/"));
            url=url.replaceAll("-", "/");
            url=url.replace(",", ".");
            url="http:/"+url;
            System.out.println(url);
            // key值由单词和URI组成。
            keyInfo.set( ch.toString()+";"+url);
            //词频初始为1
            valueInfo.set("1");
            context.write(keyInfo, valueInfo);
        }  
        ts.end();  
        ts.close();  
    }
}

